package com.atguigu.app.dwd;

import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.app.func.DimSinkFunction;
import com.atguigu.app.func.MyCustomDeserializer;
import com.atguigu.app.func.TableProcessFunction;
import com.atguigu.bean.TableProcess;
import com.atguigu.uitls.MyKafkaUtil;
import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.DebeziumSourceFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

//数据流：web/app -> nginx -> 业务服务器 -> Mysql(binlog) -> FlinkApp -> Kafka(ODS) -> FlinkApp -> Kafka/HBase
//程  序：Mock -> Mysql(binlog) -> FlinkCDC -> Kafka(ZK) ->  BaseDBApp -> Kafka/HBase(ZK,HDFS)
public class BaseDBApp {

    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);  //生产环境与Kafka主题的分区数保持一致

        //CK
        //        env.setStateBackend(new FsStateBackend("hdfs://"));
        //        env.enableCheckpointing(5000L);
        //        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //        env.getCheckpointConfig().setCheckpointTimeout(10000L);
        //        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //        env.getCheckpointConfig().setCheckpointInterval(10000L);

        //TODO 2.读取Kafka ods_base_db 主题创建主流
        DataStreamSource<String> dataStreamSource = env.addSource(MyKafkaUtil.getKafkaSource("ods_base_db", "base_db_app_210826"));

        //TODO 3.转换为JSON对象，过滤掉空数据(删除数据)
        SingleOutputStreamOperator<JSONObject> filterDS = dataStreamSource.map(JSONObject::parseObject)
                .filter(new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        return !"delete".equals(value.getString("type"));
                    }
                });

        //TODO 4.使用FlinkCDC 读取配置信息表  并创建广播流
        DebeziumSourceFunction<String> sourceFunction = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("gmall-210826-realtime")
                .tableList("gmall-210826-realtime.table_process")
                .username("root")
                .password("000000")
                .startupOptions(StartupOptions.initial())
                .deserializer(new MyCustomDeserializer())
                .build();

        MapStateDescriptor<String, TableProcess> mapStateDescriptor = new MapStateDescriptor<>("bc-state", String.class, TableProcess.class);
        BroadcastStream<String> broadcastStream = env.addSource(sourceFunction).broadcast(mapStateDescriptor);

        //TODO 5.连接主流和广播流
        BroadcastConnectedStream<JSONObject, String> connectedStream = filterDS.connect(broadcastStream);

        //TODO 6.处理广播流数据以及主流数据
        OutputTag<JSONObject> hbaseTag = new OutputTag<JSONObject>("hbase") {
        };
        SingleOutputStreamOperator<JSONObject> kafkaDS = connectedStream.process(new TableProcessFunction(mapStateDescriptor, hbaseTag));

        //TODO 7.提取侧输出流数据
        DataStream<JSONObject> hbaseDS = kafkaDS.getSideOutput(hbaseTag);

        //TODO 8.将HBase数据写出
        hbaseDS.print("HBase>>>>>>>>>");
        hbaseDS.addSink(new DimSinkFunction());

        //TODO 9.将Kafka数据写出
        kafkaDS.print("Kafka>>>>>>>>>");
        kafkaDS.addSink(MyKafkaUtil.getKafkaSink(new KafkaSerializationSchema<JSONObject>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(JSONObject element, @Nullable Long timestamp) {
                //element:{"database":"gmall-210826-flink","tableName":"order_info","before":{},"after":{"id":"...","name":"...","":"..."},"type":"insert","sinkTable":"dwd_order_info"}
                return new ProducerRecord<>(element.getString("sinkTable"),
                        element.getString("after").getBytes());
            }
        }));

        //TODO 10.启动任务
        env.execute("BaseDBApp");
    }

}
